package spring.cloud.tasks.client;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationContext;
import spring.cloud.tasks.client.item_task_processor.*;
import spring.cloud.tasks.common.utils.ApplicationContextUtils;
import spring.cloud.tasks.common.zookeeper.ZooKeeperDataManager;
import spring.cloud.tasks.common.zookeeper.ZooKeeperItemTaskDataManager;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;


@Slf4j
public class ItemTaskWorker extends Worker {


    @Setter
    @Getter
    private ZooKeeperItemTaskDataManager zooKeeperItemTaskDataManager;


    @Setter
    @Getter
    private Map<String, List<TaskProcessor>> taskIdToTaskProcessorListMap = new ConcurrentHashMap<String, List<TaskProcessor>>();
    @Setter
    @Getter
    private Map<String, Integer> taskIdToTaskItemVersionMap = new ConcurrentHashMap<String, Integer>();


    public ItemTaskWorker() {
        try {
            this.ip = InetAddress.getLocalHost().getHostAddress();
        } catch (UnknownHostException e) {
            e.printStackTrace();
        }

        ApplicationContext applicationContext = ApplicationContextUtils.getApplicationContext();
        Map<String, spring.cloud.tasks.client.item_task_processor.api.TaskProcessor> stringTaskDataProcessorMap = applicationContext.getBeansOfType(spring.cloud.tasks.client.item_task_processor.api.TaskProcessor.class);
        for (spring.cloud.tasks.client.item_task_processor.api.TaskProcessor taskProcessor : stringTaskDataProcessorMap.values()) {
            taskIdList.add(taskProcessor.getTaskId());
        }
    }

    public void initZooKeeperDataManager() throws Exception {
        zooKeeperItemTaskDataManager = new ZooKeeperItemTaskDataManager(zooKeeperManager);
    }

    public void cleanZooKeeperDataManager() {
        zooKeeperItemTaskDataManager = null;
    }

    @Override
    public ZooKeeperDataManager getZooKeeperDataManager() {
        return zooKeeperItemTaskDataManager;
    }

    public void stopAll() throws Exception {
        stopTaskProcessorList();
    }


    public void pongpong() throws Exception {
        this.lock.lock();
        try {
            Set<String> taskIdSet = this.taskIdToTaskProcessorListMap.keySet();
            for (String taskId : taskIdSet) {
                List<TaskProcessor> taskProcessorList = this.taskIdToTaskProcessorListMap.get(taskId);
                List<TaskProcessor> needRemoveTaskProcessorList = new ArrayList<TaskProcessor>();
                for (TaskProcessor taskProcessor : taskProcessorList) {
                    if (taskProcessor.isExceptionOccurred()) {
                        needRemoveTaskProcessorList.add(taskProcessor);
                    }
                }
                taskProcessorList.removeAll(needRemoveTaskProcessorList);
                for (TaskProcessor taskProcessor : needRemoveTaskProcessorList) {
                    taskProcessor.stop();
                }
            }

        } finally {
            this.lock.unlock();
        }
        this.lock.lock();
        try {
            WorkerInfo workerInfo = null;
            boolean fileNotFoundException = false;
            try {
                workerInfo = this.getZooKeeperItemTaskDataManager().loadWorkerInfo(this.getWorkerId());//总体信息
            } catch (Exception e) {
                log.error("获取workerInfo失败", e);
                fileNotFoundException = true;
            }
            if (fileNotFoundException == true) {//被清理了
                try {
                    stopTaskProcessorList();
                    this.getZooKeeperItemTaskDataManager().unRegisterWorker(this);//卸载调度器
                    //

                } finally {
                    rebuildTaskProcessorListWhenAllowExecuteTasks();
                }
            } else {
                if (workerInfo.isExecuteTasks() == false) {
                    stopTaskProcessorList();

                } else {
                    //正常
                    rebuildTaskProcessorListWhenAllowExecuteTasks();

                }
            }
        } finally {
            this.lock.unlock();
        }
    }


    public void rebuildTaskProcessorListWhenAllowExecuteTasks() throws Exception {
        //只支持worker允许运行的情况下才进行调用
        List<String> taskIdList = this.getZooKeeperItemTaskDataManager().registerWorkerToPermittedTaskListWhenPongPong(this);
        //这个是允许存活的任务,排开这些之外的需要进行清理
        this.stopTaskProcessorListNotTaskIdList(taskIdList);
        this.rebuildTaskProcessorListByWorkerId(taskIdList);
    }


    public TaskProcessor createItemTaskProcessorInfo(Task task) throws Exception {
        TaskProcessor taskProcessor = null;
        try {
            taskProcessor = new TaskProcessor(task.getTaskId(), this, zooKeeperItemTaskDataManager, task.getMinThreadCountPerProcessor());
        } catch (Exception e) {
            log.error("schedule" + task.getTaskId(), e);
        }
        return taskProcessor;
    }

    public void stopTaskProcessorList() throws Exception {
        String[] taskIdList = this.taskIdToTaskProcessorListMap.keySet().toArray(new String[0]);
        for (String taskId : taskIdList) {
            for (TaskProcessor taskProcessor : this.taskIdToTaskProcessorListMap.get(taskId)) {
                try {
                    if (taskProcessor != null) {
                        taskProcessor.stop();
                    }
                } catch (Throwable e) {
                    log.error("id=" + taskId, e);
                }
            }
            this.taskIdToTaskProcessorListMap.remove(taskId);
        }
    }

    public void stopTaskProcessorListNotTaskIdList(List<String> taskIdList) throws Exception {
        //找到所有的taskIdList
        List<String> needRemoveTaskIdList = new ArrayList<String>(this.taskIdToTaskProcessorListMap.keySet());
        needRemoveTaskIdList.removeAll(taskIdList);
        for (String taskId : needRemoveTaskIdList) {
            stopTaskProcessorList(taskId);
        }

    }

    public void stopTaskProcessorList(String taskId) throws Exception {
        List<TaskProcessor> taskProcessorList = this.taskIdToTaskProcessorListMap.get(taskId);
        if (taskProcessorList != null) {
            for (TaskProcessor taskProcessor : taskProcessorList) {
                try {
                    if (taskProcessor != null) {
                        taskProcessor.stop();
                    }
                } catch (Throwable e) {
                    log.error("id=" + taskId, e);
                }
            }
            this.taskIdToTaskProcessorListMap.remove(taskId);
        }
    }

    public void rebuildTaskProcessorListByWorkerId(List<String> taskIdList) throws Exception {
        //先进行taskItem状态监听
        for (String taskId : taskIdList) {
            //版本号不一致 则直接清除。如果新的版本号出于分配中则什么也不进行处理。
            TaskItemsInfo taskItemInfo = zooKeeperItemTaskDataManager.getTaskItemInfo(taskId);
            if (!taskIdToTaskItemVersionMap.containsKey(taskId)) {
                taskIdToTaskItemVersionMap.put(taskId, taskItemInfo.getVersion());
                continue;
            } else {
                int version = taskIdToTaskItemVersionMap.get(taskId);
                if (taskItemInfo.getVersion() > version) {
                    List<TaskProcessor> taskProcessorList = this.taskIdToTaskProcessorListMap.remove(taskId);
                    for (TaskProcessor taskProcessor : taskProcessorList) {
                        try {
                            taskProcessor.stop();//等下个调度时间 会重新根据task能力(因为版本回退可能减少)+itemTaskWorker+ip再重新进行调度器分配。
                        } catch (Throwable e) {
                            log.error(taskId, e);
                        }
                    }
                    taskIdToTaskItemVersionMap.put(taskId, taskItemInfo.getVersion());
                }

            }
        }
        //只要task item 版本号改变。
        //只允许合法的taskIdList
        List<TaskWorkerScheduleInfo> taskWorkerScheduleInfoList = this.getZooKeeperItemTaskDataManager().loadTaskWorkerScheduleInfoListByWorkerId(this.workerId);
        for (TaskWorkerScheduleInfo taskWorkerScheduleInfo : taskWorkerScheduleInfoList) {
            String taskId = taskWorkerScheduleInfo.getTaskId();
            Task task = getZooKeeperItemTaskDataManager().loadTask(taskId);
            //
            this.taskIdToTaskProcessorListMap.putIfAbsent(taskId, new ArrayList<TaskProcessor>());
            List<TaskProcessor> taskProcessorList = this.taskIdToTaskProcessorListMap.get(taskId);
            //
            if (!taskIdList.contains(taskId)) {
                //调取不允许该IP或者该worker执行
                for (TaskProcessor taskProcessor : taskProcessorList) {
                    try {
                        taskProcessor.stop();//等下个调度时间 会重新根据task能力(因为版本回退可能减少)+itemTaskWorker+ip再重新进行调度器分配。
                    } catch (Throwable e) {
                        log.error(taskId, e);
                    }
                }
            } else {
                TaskItemsInfo taskItemInfo = zooKeeperItemTaskDataManager.getTaskItemInfo(taskId);
                if (taskItemInfo.getVersion() > taskWorkerScheduleInfo.getTaskItemVserion()) {
                    for (TaskProcessor taskProcessor : taskProcessorList) {
                        try {
                            taskProcessor.stop();
                        } catch (Throwable e) {
                            log.error(taskId, e);
                        }
                    }
                    //后加
                    taskProcessorList.clear();
                    int requestNum = taskWorkerScheduleInfo.getRequestNum();
                    while (taskProcessorList.size() < requestNum) {
                        TaskProcessor taskProcessor = this.createItemTaskProcessorInfo(task);
                        taskProcessorList.add(taskProcessor);
                    }
                    continue;//需要重新进行分配。
                } else {
                    //先清理task items过期的处理器
                    for (TaskProcessor taskProcessor : taskProcessorList) {
                        TaskProcessorInfo taskProcessorInfo = zooKeeperItemTaskDataManager.getProcessorInfo(taskId, taskProcessor.getTaskProcessorInfo().processorId);
                        if (taskProcessorInfo == null) {
                            taskProcessorList.remove(taskProcessor);
                            try {
                                taskProcessor.stop();
                            } catch (Throwable e) {
                                log.error(taskId, e);
                            }
                        } else {
                            String needDestroyWhenTaskItemsInvalid = taskProcessorInfo.needDestroyWhenTaskItemsInvalid;
                            if ("Y".equals(needDestroyWhenTaskItemsInvalid)) {
                                taskProcessorList.remove(taskProcessor);
                                try {
                                    taskProcessor.stop();
                                } catch (Throwable e) {
                                    log.error(taskId, e);
                                }
                            }
                        }
                    }
                    int requestNum = taskWorkerScheduleInfo.getRequestNum();
                    //先减
                    while (taskProcessorList.size() > 0 && taskProcessorList.size() > requestNum) {
                        TaskProcessor taskProcessor = taskProcessorList.remove(taskProcessorList.size() - 1);
                        try {
                            taskProcessor.stop();
                        } catch (Throwable e) {
                            log.error(taskId, e);
                        }
                    }
                    //调度器线程数分配
                    List<TaskProcessor> newTaskProcessorList = new ArrayList<>();
                    while (taskProcessorList.size() > 0) {
                        TaskProcessor taskProcessor = taskProcessorList.remove(taskProcessorList.size() - 1);
                        int threadCount = taskProcessor.getThreadCountPerTaskProcessor();
                        if (threadCount != task.getMinThreadCountPerProcessor()) {
                            try {
                                taskProcessor.stop();
                            } catch (Throwable e) {
                                log.error(taskId, e);
                            }
                            taskProcessor = this.createItemTaskProcessorInfo(task);
                        }
                        newTaskProcessorList.add(taskProcessor);
                    }
                    taskProcessorList.addAll(newTaskProcessorList);
                    //后加
                    while (taskProcessorList.size() < requestNum) {
                        TaskProcessor taskProcessor = this.createItemTaskProcessorInfo(task);
                        taskProcessorList.add(taskProcessor);
                    }
                }
            }

        }
    }


}

